package com.lwhtarena.netty.netty4.util;

import io.debezium.connector.mysql.MySqlConnector;
import io.debezium.embedded.Connect;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import io.debezium.relational.history.FileDatabaseHistory;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.storage.FileOffsetBackingStore;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class DebeziumAutoConfig {


    private String mysqlHostName;

    private String mysqlPort;

    private String mysqlUser;

    private String mysqlPassword;

    public static void main(String[] args) {
        // 配置 Debezium
        Properties props = new Properties();


        props.setProperty("name", "mysql-connector");
        props.setProperty("key.converter.schemas.enable", "false");
        props.setProperty("value.converter.schemas.enable", "false");
        props.setProperty("include.schema.changes", "false");
        props.setProperty("tombstones.on.delete", "false");
        props.setProperty("include.query", "false");
        props.setProperty("connector.class", MySqlConnector.class.getCanonicalName());
        props.setProperty("offset.storage", FileOffsetBackingStore.class.getName());
        props.setProperty("offset.storage.file.filename", "./workes/hengjin-weight/working-items/offset-file.dat"); // 存储偏移量的文件路径
        props.setProperty("offset.flush.interval.ms", "60000");
        props.setProperty("database.hostname", "192.168.2.13");
        props.setProperty("database.port", "3306");
        props.setProperty("database.user", "root");
        props.setProperty("database.password", "root");
        props.setProperty("database.server.id", "12345"); // 唯一的 server id
        props.setProperty("database.server.name", "12345");
        props.setProperty("database.include.list", "hengjin");
       // props.setProperty("table.include.list", "test.stu");
        props.setProperty("database.history.instance.name", UUID.randomUUID().toString());
        props.setProperty("database.history.skip.unparseable.ddl", "true");
        props.setProperty("database.history.store.only.monitored.tables.ddl", "true");

        props.setProperty("database.history",
                FileDatabaseHistory.class.getCanonicalName());
        props.setProperty("database.history.file.filename",
                "./workes/hengjin-weight/working-items/dbhistory.dat");


        // 创建 Debezium 引擎

        try{
            DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
                    .using(props)
                    .notifying((record,committer) -> {
                        // record中会有操作的类型（增、删、改）和具体的数据
                        // key是主键
                        for (ChangeEvent<String,String> record1:record) {
                            System.out.println("record.key() = " + record1.key());
                            System.out.println("record.value() = " + record1.value());
                        }
                       committer.markBatchFinished();
                    })
                    .using((success, message, error) -> {
                        // 强烈建议加上此部分的回调代码，方便查看错误信息
                        if (error != null){
                            error.printStackTrace();
                        }
                        System.out.println("success = " + success);
                        System.out.println("message = " + message);
                        System.out.println("error = " + error);
                    }).build();
            // 启动 Debezium 引擎
            DebeziumEngine<ChangeEvent<SourceRecord,SourceRecord>> recordEngine = DebeziumEngine.create(Connect.class)
                    .using(props)
                    .notifying((record,committer) -> {
                        // record中会有操作的类型（增、删、改）和具体的数据
                        // key是主键
                        for (ChangeEvent<SourceRecord,SourceRecord> record1:record
                        ) {
                            System.out.println("record.key() = " + record1.key());
                            System.out.println("record.value() = " + record1.value());
                        }
                        committer.markBatchFinished();
                    }).build();
            ExecutorService executor = Executors.newSingleThreadExecutor();
            executor.execute(engine);
            // 等待 10 秒钟
            //TimeUnit.SECONDS.sleep(1000);
            // 停止 Debezium 引擎
//            engine.close();
//            executor.shutdown();
        }catch (Exception e){
            e.printStackTrace();
            System.out.println(e.getMessage());
        }


    }


}
